/**
  ETFAna project, Anyang Normal University && IMP-CAS
  \class ETFOnlineSource
  \brief to process data stream from daq while it's working, i.e., during the
  experiment
  \author SUN Yazhou, asia.rabbit@163.com
  \since 2021-09-28
  \date 2024-06-26 last modified
  \attention
  changelog
  <table>
  <tr>  <th>Date        <th>Author      <th>Description                    </tr>
  <tr>  <td>2021-09-28   <td>Asia Sun    <td>file created                  </tr>
  </table>

  \copyright Copyright (c) 2021-2024 Anyang Normal U. && IMP-CAS with LGPLv3 LICENSE
*/

#include <cstring>
#include <unistd.h>
#include <iostream>
#include "ETFOnlineSource.h"
#include "ETFPXIUnpacker.h"
#include "ETFVMEUnpacker.h"
#include "ETFVClient.h"
#include "ETFMsg.h"
#include "ETFMath.h"
#include "ETFUnpacker.h"
#include "ETFQueue.h"
#include "ETFOnlineCache.h"

#define er ETFMsg::Error

// #define SYNC_DAQ // just for sync daqs - use with CAUTION

using std::cout;
using std::endl;

ETFOnlineSource::ETFOnlineSource(const string &name) : ETFSource(), fDropped(0),
fIsClean(true), fSkipPXI(false), fSkipVME(false){
  fROOTFileName = name + ETFMsg::time0(1) + ".root";
  fCache = new ETFOnlineCache();
} // end ctor

ETFOnlineSource::~ETFOnlineSource(){
  Close();
} // end dtor

// add a daq source
void ETFOnlineSource::AddDaqSource(ETFDaqData *q){
  if(!q) er("ETFOnlineSource", "AddUnpacker: nullptr daq data encountered.");
  const DaqType t = q->daqType();
  int id = 0;
  for(const auto &p : fCache->daqs()) if(p->daq->daqType() == t) id++;
  ETFUnpacker *u = nullptr;
  if(t == kPXI) u = new ETFPXIUnpacker(id);
  else u = new ETFVMEUnpacker(id);

  fCache->AddDaq(u, q);
  fUpks.push_back(u);
} // end member function AddUnpacker

// initiate the reading threads
void ETFOnlineSource::Online(){
  if(!fUpks.size()) er("ETFOnlineSource", "Init: no unpacker found.");
  // set all daq sources online //
  for(const auto &p : fCache->daqs()) p->daq->Initialize();
} // end member function Online

// \retval true: success
bool ETFOnlineSource::ReadEvent(unsigned){
  if(!fCache->GetOneEvent()) return false; // the queues are empty or events can not be matched

  int i = 0;
  for(auto &d : fCache->daqs()){
    auto u = d->upk;
    u->Reset();
    u->GetOneEvent(d->cache[fCache->MatchedId()[i++]]);
    u->Unpack();
  } // end for over upks

  fEventCnt = fCache->CntEv();
  return true;
} // end member function ReadEvent

// true if at least one daq srcs is empty (loaded < nm)
bool ETFOnlineSource::empty(int nm) const{
  bool res = false;
  for(const auto &p : fCache->daqs()) if(p->daq->queue()->NLoaded() < nm) res = true;
  return res;
} // end member function empty

// close the daq data object
void ETFOnlineSource::Close(){
  for(const auto &p : fCache->daqs()){
    delete p->daq;
    delete p->upk;
  } // end for

  // remove the generated rootfile on demand
  const char *name = ROOTFileName();
  if(fIsClean && !access(name, F_OK)) if(-1 == remove(name))
    er("ETFOnlineSource", "dtor: remove file %s error: %s",
      name, strerror(errno));
} // end member function Close

// clear all the daqs and data
void ETFOnlineSource::Clear(){
  fCache->Clear();
} // end member function Clear

int ETFOnlineSource::CacheLoaded() const{
  return fCache->loaded();
} // end member function CacheLoaded
